package com.xln.kafka.sample.sample;




import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class ProducerQuickStart {


    public static void main(String[] args) {


        Properties properties = new Properties();
        properties.put("bootstrap.servers","192.168.200.130:9092");
        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

        properties.put("acks","1");
        properties.put("retries",10);
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

        KafkaProducer producer   = new KafkaProducer(properties);
//        ProducerRecord<String, String> record = new ProducerRecord<>("topic-first", "key-001", "hello kafka");
//        producer.send(new ProducerRecord< String,String>("topic-first","key-001","hello kafka"));

        //同步发送
//        try {
//            RecordMetadata recordMetadata =(RecordMetadata) producer.send(record).get();
//            System.out.println("发送成功.主题:"+recordMetadata.topic()+"分区:"+recordMetadata.partition()+"偏移:"+recordMetadata.offset());
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        } catch (ExecutionException e) {
//            throw new RuntimeException(e);
//        }

        //异步发送
//        producer.send( record,new Callback() {
//
//            @Override
//            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
//                if(e == null){
//                    System.out.println("发送成共.主题:"+recordMetadata.topic()+"分区:"+recordMetadata.partition()+"偏移:"+recordMetadata.offset());
//                }else{
//                    System.out.println("发送失败");
//                    e.printStackTrace();
//                }
//            }
//        });

        //stream 方式发送
        for (int i = 0; i < 5; i++) {
            ProducerRecord<String,String> kvProducerRecord = new ProducerRecord<String,String>("stream-quickstart-input","hello kafka");
            producer.send(kvProducerRecord);
        }


        producer.close();
    }

}
